package juc;

import java.util.Date;
import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.stream.IntStream;


/**
 * 测试 java.util.concurrent.Exchanger
 * 两个线程之间交换数据，
 * 调用其exchange方法交换，
 * 首先调用exchange的等待
 * exchange有超时版本：时间到后，对手线程没有调用exchange，本线程发生TimeOutException，
 * 此时，不再次执行exchange，则另一个线程将会【一直等待】！
 * @author ben
 * @date 2021-09-19 22:08:03 CST
 */
public class TestExchanger {

	public static Consumer<Object> cs = System.out::println;

	public static void main(String[] args) {
//		test1(1);
//		test1(2);
		
		test2();
	}
	
	/**
	 * 主线程和非主线程交换数据
	 * @author ben
	 * @date 2021-09-19 22:10:55 CST
	 * @param v 1-exchange正常版本，2-exchange超时版本
	 */
	public static void test1(int v) {
		final String tnameMain = Thread.currentThread().getName();
		
		cs.accept(tnameMain + ": test1 start...v=" + v);
		Exchanger<String> ex = new Exchanger<>();
		cs.accept("ex=" + ex);
		
		new Thread(()->{
			final String tname = Thread.currentThread().getName();
			cs.accept(tname + ": start...now=" + new Date());
			try {
				TimeUnit.SECONDS.sleep(21);
				cs.accept(tname + ": after sleep...now=" + new Date());
				String fr = ex.exchange(Thread.currentThread().getName());
				cs.accept(tname + ": fr=" + fr + ", now=" + new Date());
			} catch (InterruptedException e) {
				cs.accept(tname + ": e=" + e.getClass());
			}
			
			cs.accept(tname + ": end.");
		}).start();
		
		try {
			cs.accept(tnameMain + ": beforeexchange...now=" + new Date());
			String fr = "";
			if (v == 1) {
				// 1、正常版本
				fr = ex.exchange(Thread.currentThread().getName());
			} else if (v == 2) {
				// 2、超时版本
				// 超时需处理，否则，对手线程会一直阻塞、本程序也不会停止！
				while (true) {
					try {
						fr = ex.exchange(Thread.currentThread().getName(), 5L, TimeUnit.SECONDS);
						// 等到，跳出循环
						break;
					} catch (TimeoutException e) {
						cs.accept(tnameMain + ": exchange e=" + e.getClass() + ", now=" + new Date());
					}
				}
			}
			cs.accept(tnameMain + ": fr=" + fr + ", now=" + new Date());
		} catch (InterruptedException e) {
			cs.accept(tnameMain + ": 1 e=" + e.getClass());
//		} catch (TimeoutException e) {
//			// 移动到exchange的地方
//			cs.accept(tnameMain + ": 2 e=" + e.getClass());
		} finally {
			cs.accept(tnameMain + ": finally now=" + new Date());
		}
		
		cs.accept(tnameMain + ": test1 end.");
	}
	
	/**
	 * 两个非主线程不停执行exchange
	 * @author ben
	 * @date 2021-09-19 22:52:12 CST
	 */
	public static void test2() {
		String tnameMain = Thread.currentThread().getName();
		cs.accept(tnameMain + ": test2 start...");

		Exchanger<String> ex = new Exchanger<>();
		cs.accept("ex=" + ex);
		
		Thread t1 = new Thread(()->{
			String tname = Thread.currentThread().getName();
			cs.accept(tname + ": start...now=" + new Date());
			
			IntStream.range(0, 10).forEach(item->{
				try {
					TimeUnit.SECONDS.sleep(2);
					
					String fr = ex.exchange(tname + "-" + item);
					cs.accept(tname + ": fr=" + fr + ", now=" + new Date());
				} catch (InterruptedException e) {
					cs.accept(tname + ": e=" + e.getClass());
				}
			});
			
			cs.accept(tname + ": end...now=" + new Date());
		}, "T1");

		
		Thread t2 = new Thread(()->{
			String tname = Thread.currentThread().getName();
			cs.accept(tname + ": start...now=" + new Date());
			
			IntStream.range(100, 110).forEach(item->{
				try {
					TimeUnit.SECONDS.sleep(4);
					
					String fr = ex.exchange(tname + "-" + item);
					cs.accept(tname + ": fr=" + fr + ", now=" + new Date());
				} catch (InterruptedException e) {
					cs.accept(tname + ": e=" + e.getClass());
				}
			});
			
			cs.accept(tname + ": end...now=" + new Date());
		}, "T2");
		
		t1.start();
		t2.start();
		
		cs.accept(tnameMain + ": test2 end.");
	}

}
